package com.atguigu.realtime.app.dwd;

import com.atguigu.realtime.app.BaseSQLApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dwd_PaymentInfo extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_PaymentInfo().init(
                4003,
                2,
                "Dwd_PaymentInfo"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv,"Dwd_PaymentInfo");

        Table paymentInfo = tEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['out_trade_no'] out_trade_no, " +
                " `data`['order_id'] order_id, " +
                " `data`['alipay_trade_no'] alipay_trade_no, " +
                " `data`['total_amount'] total_amount," +
                " `data`['trade_body'] trade_body," +
                " `data`['payment_type'] payment_type," +
                " `data`['payment_status'] payment_status," +
                " `data`['create_time'] create_time," +
                " `data`['update_time'] update_time," +
                " `data`['callback_content'] callback_content," +
                " `data`['callback_time'] callback_time " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='payment_info' ");

        tEnv.createTemporaryView("payment_info",paymentInfo);

        tEnv.executeSql("create table dwd_payment_info( " +
                "id string, " +
                "out_trade_no string, " +
                "order_id string, " +
                "alipay_trade_no string, " +
                "total_amount string," +
                "trade_body string," +
                "payment_type string," +
                "payment_status string," +
                "create_time string," +
                "update_time string," +
                "callback_content string," +
                "callback_time string " +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_PAYMENT_INFO));

        paymentInfo.executeInsert("dwd_payment_info");

    }

}
